using Confluent.Kafka;
using CoreManager.MessageQueue.Models;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Newtonsoft.Json;

namespace CoreManager.MessageQueue.Services;

/// <summary>
/// Kafka消费者服务实现
/// </summary>
public class KafkaConsumerService : IKafkaConsumerService, IDisposable
{
    private readonly IConsumer<string, string> _consumer;
    private readonly ILogger<KafkaConsumerService> _logger;
    private readonly KafkaOptions _options;
    private bool _disposed;
    private CancellationTokenSource? _cancellationTokenSource;

    public KafkaConsumerService(IOptions<KafkaOptions> options, ILogger<KafkaConsumerService> logger)
    {
        _options = options.Value;
        _logger = logger;

        var config = new ConsumerConfig
        {
            BootstrapServers = _options.BootstrapServers,
            GroupId = _options.Consumer.GroupId,
            EnableAutoCommit = _options.Consumer.EnableAutoCommit,
            AutoCommitIntervalMs = _options.Consumer.AutoCommitIntervalMs,
            AutoOffsetReset = Enum.Parse<AutoOffsetReset>(_options.Consumer.AutoOffsetReset, true),
            EnablePartitionEof = false,
            SessionTimeoutMs = 30000,
            HeartbeatIntervalMs = 3000
        };

        _consumer = new ConsumerBuilder<string, string>(config)
            .SetErrorHandler((_, error) =>
            {
                _logger.LogError("Kafka consumer error: {Error}", error.Reason);
            })
            .SetLogHandler((_, log) =>
            {
                _logger.LogInformation("Kafka consumer log: {Message}", log.Message);
            })
            .SetPartitionsAssignedHandler((consumer, partitions) =>
            {
                _logger.LogInformation("Partitions assigned: {Partitions}", 
                    string.Join(", ", partitions.Select(p => $"{p.Topic}[{p.Partition}]")));
            })
            .SetPartitionsRevokedHandler((consumer, partitions) =>
            {
                _logger.LogInformation("Partitions revoked: {Partitions}", 
                    string.Join(", ", partitions.Select(p => $"{p.Topic}[{p.Partition}]")));
            })
            .Build();

        _logger.LogInformation("Kafka consumer initialized with servers: {Servers}, GroupId: {GroupId}", 
            _options.BootstrapServers, _options.Consumer.GroupId);
    }

    /// <inheritdoc />
    public void Subscribe(params string[] topics)
    {
        Subscribe(topics.AsEnumerable());
    }

    /// <inheritdoc />
    public void Subscribe(IEnumerable<string> topics)
    {
        ThrowIfDisposed();
        
        var topicList = topics.ToList();
        _consumer.Subscribe(topicList);
        _logger.LogInformation("Subscribed to topics: {Topics}", string.Join(", ", topicList));
    }

    /// <inheritdoc />
    public async Task StartConsumingAsync(Func<string, KafkaMessage, Task<bool>> messageHandler, CancellationToken cancellationToken = default)
    {
        ThrowIfDisposed();

        _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        var token = _cancellationTokenSource.Token;

        _logger.LogInformation("Starting Kafka message consumption");

        try
        {
            await Task.Run(async () =>
            {
                while (!token.IsCancellationRequested)
                {
                    try
                    {
                        var consumeResult = _consumer.Consume(token);
                        
                        if (consumeResult?.Message?.Value != null)
                        {
                            var message = JsonConvert.DeserializeObject<KafkaMessage>(consumeResult.Message.Value);
                            if (message != null)
                            {
                                _logger.LogDebug("Received message from topic {Topic}, partition {Partition}, offset {Offset}", 
                                    consumeResult.Topic, consumeResult.Partition.Value, consumeResult.Offset.Value);

                                var success = await messageHandler(consumeResult.Topic, message);
                                
                                if (success)
                                {
                                    if (!_options.Consumer.EnableAutoCommit)
                                    {
                                        _consumer.Commit(consumeResult);
                                    }
                                    _logger.LogDebug("Message processed successfully");
                                }
                                else
                                {
                                    _logger.LogWarning("Message processing failed for message {MessageId}", message.MessageId);
                                }
                            }
                        }
                    }
                    catch (ConsumeException ex)
                    {
                        _logger.LogError(ex, "Error consuming message");
                    }
                    catch (OperationCanceledException)
                    {
                        break;
                    }
                    catch (Exception ex)
                    {
                        _logger.LogError(ex, "Unexpected error during message consumption");
                    }
                }
            }, token);
        }
        finally
        {
            _logger.LogInformation("Kafka message consumption stopped");
        }
    }

    /// <inheritdoc />
    public async Task StartConsumingAsync<T>(Func<string, KafkaMessage<T>, Task<bool>> messageHandler, CancellationToken cancellationToken = default)
    {
        await StartConsumingAsync(async (topic, message) =>
        {
            if (message.Data is T typedData)
            {
                var typedMessage = new KafkaMessage<T>
                {
                    MessageId = message.MessageId,
                    MessageType = message.MessageType,
                    Timestamp = message.Timestamp,
                    Data = typedData,
                    Source = message.Source,
                    Version = message.Version
                };
                
                return await messageHandler(topic, typedMessage);
            }
            
            _logger.LogWarning("Message data type mismatch. Expected {ExpectedType}, got {ActualType}", 
                typeof(T).Name, message.Data?.GetType().Name ?? "null");
            return false;
        }, cancellationToken);
    }

    /// <inheritdoc />
    public void StopConsuming()
    {
        _cancellationTokenSource?.Cancel();
        _logger.LogInformation("Kafka message consumption stop requested");
    }

    /// <inheritdoc />
    public void CommitOffset()
    {
        ThrowIfDisposed();
        _consumer.Commit();
        _logger.LogDebug("Manual offset commit performed");
    }

    /// <inheritdoc />
    public void Dispose()
    {
        if (_disposed) return;

        try
        {
            _cancellationTokenSource?.Cancel();
            _cancellationTokenSource?.Dispose();
            _consumer?.Close();
            _consumer?.Dispose();
            _logger.LogInformation("Kafka consumer disposed");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error disposing Kafka consumer");
        }
        finally
        {
            _disposed = true;
        }
    }

    private void ThrowIfDisposed()
    {
        if (_disposed)
        {
            throw new ObjectDisposedException(nameof(KafkaConsumerService));
        }
    }
}
